package com.bw.gmall.realtime.dwd.db.app;

import com.bw.gmall.realtime.common.base.BaseSQLApp;
import com.bw.gmall.realtime.common.constant.Constant;
import com.bw.gmall.realtime.common.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdTradeCartAdd extends BaseSQLApp {
    public static void main(String[] args) {
        new DwdTradeCartAdd().start(10013,4, Constant.TOPIC_DWD_TRADE_CART_ADD);
    }
    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tabEnv) {
        //TODO 从kafka中的topic_db 主题中读取数据 创建动态表
        readOdsDb(tabEnv,Constant.TOPIC_DWD_TRADE_CART_ADD);
        //TODO 过滤出加购数据 table 'cart_info' type = 'insert' ,, type = 'update' 并且修改的是加购商品的数量,修改后的值大于修改前的值



        Table cartAdd = tabEnv.sqlQuery("select " +
                " `data`['id'] id," +
                " `data`['user_id'] user_id," +
                " `data`['sku_id'] sku_id," +
                " if(`type`='insert'," +
                " cast(`data`['sku_num'] as int), " +
                " cast(`data`['sku_num'] as int) - cast(`old`['sku_num'] as int)" + ") sku_num ," +
                " ts " +
                "from topic_db " +
                "where `database`='gmall' " +
                "and `table`='cart_info' " +
                "and (" +
                " `type`='insert' " +
                " or ( " +
                " `type`='update' " +
                " and `old`['sku_num'] is not null " +
                " and cast(`data`['sku_num'] as int) > cast(`old`['sku_num'] as int) " + // Cast('' as int ) FlinkSql中的转换函数  指定类型的转换函数
                " )" +
                ")");
//        cartAdd.execute().print();
        //TODO 将过滤出来的 加购数据写到kafka主题中
        //创建动态表和写入的主题进行映射
        tabEnv.executeSql("create table "+Constant.TOPIC_DWD_TRADE_CART_ADD+"(" +
                " id string, " +
                " user_id string," +
                " sku_id string," +
                " sku_num int, " +
                " ts bigint " +
                ")" +
                SQLUtil.getKafkaDDLSink(Constant.TOPIC_DWD_TRADE_CART_ADD));
        //写入
        cartAdd.executeInsert(Constant.TOPIC_DWD_TRADE_CART_ADD);
    }
}
